/*
 * (c) Copyright 2009 Hewlett-Packard Development Company, LP
 * All rights reserved.
 * [See end of file]
 */

package com.hp.hpl.jena.clusteredtdb.pgraph;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;

import com.hp.hpl.jena.clusteredtdb.base.UtilityFunctions;
import com.hp.hpl.jena.clusteredtdb.comms.ClusterRepresentation;
import com.hp.hpl.jena.clusteredtdb.comms.NullProcessor;
import com.hp.hpl.jena.clusteredtdb.comms.RecordProcessor;
import com.hp.hpl.jena.clusteredtdb.comms.RemoteVNode;
import com.hp.hpl.jena.clusteredtdb.comms.Request;
import com.hp.hpl.jena.clusteredtdb.comms.StringProcessor;
import com.hp.hpl.jena.tdb.base.record.Record;
import com.hp.hpl.jena.tdb.base.record.RecordFactory;

public class NetworkedRangeIndex implements BulkRangeIndex, Iterable<Record> {

	protected ClusterRepresentation cr;
	protected RecordFactory rf;
	protected String indexName;
	
	/*
	 * Need to provide some means of telling the remote server
	 * what index we want to hit.  Could do this via port numbers,
	 * but for now just giving the index a unified name
	 */
	public NetworkedRangeIndex(ClusterRepresentation cr) {
		this(cr, Names2.indexSPO);
	}
	
	public NetworkedRangeIndex(ClusterRepresentation cr, String indexName) {
		this.cr = cr;
		rf = new RecordFactory(24,24);
		this.indexName = indexName;
	}
	
	/*
	 * Selects a vnode based on key.  Random for now, should be replaced
	 * with hash keys mod numvnodes soon.
	 * 
	 * NB hash function could be faster if we use a power of 2 # of vnodes
	 * combined with shifting, rather than the modulus.
	 */
	protected RemoteVNode findVNode(byte[] key) {
		//hash the first 8 bytes of the key
		int hash = UtilityFunctions.hashLong(ByteBuffer.wrap(key).asLongBuffer().get(0));
		return cr.getVNodeForId(Math.abs(hash % cr.getVNodeCount()));
	}
	
	
	/*
	 * TODO returns an iterator for all data.  Needs sorting?
	 * @see com.hp.hplNames.indexSPO.jena.tdb.index.RangeIndex#iterator()
	 */
	public Iterator<Record> iterator() {
		// TODO Auto-generated method stub
		return null;
	}

	/*
	 * TODO returns an iterator over a range of records.
	 * Will not initially work over many S.
	 * @see com.hp.hpl.jena.tdb.index.RangeIndex#iterator(com.hp.hpl.jena.tdb.base.record.Record, com.hp.hpl.jena.tdb.base.record.Record)
	 */
	public Iterator<Record> iterator(Record recordMin, Record recordMax) {
		RemoteVNode vnode = this.findVNode(recordMin.getKey());
		
		RecordProcessor rp = new RecordProcessor();
		ArrayList<String> ar = new ArrayList<String>();
		ar.add("FINDRANGE");
		ar.add(indexName);
		String str1 = UtilityFunctions.createSpacedTupleString(
				UtilityFunctions.tupleBytesToStrings(recordMin.getKey()));
		String str2 = UtilityFunctions.createSpacedTupleString(
				UtilityFunctions.tupleBytesToStrings(recordMax.getKey()));

		ar.add(str1);
		ar.add(str2);

		rp = (RecordProcessor)cr.SendOneVNode(new Request(vnode,ar, false),rp);
		
		Iterator<Record> tupleIterator = rp.getData();
				
		return tupleIterator;
	
	}

	/*
	 * Need to have a system for recording which system has the max key,
	 * otherwise this will be very inefficient.
	 * @see com.hp.hpl.jena.tdb.index.RangeIndex#maxKey()
	 */
	public Record maxKey() {
		// TODO Auto-generated method stub
		return null;
	}

	/*
	 * Need to have a system for recording which system has the min key,
	 * otherwise this will be very inefficient. 
	 * @see com.hp.hpl.jena.tdb.index.RangeIndex#minKey()
	 */
	public Record minKey() {
		// TODO Auto-generated method stub
		return null;
	}

	public boolean add(Record record) {
		RemoteVNode vnode = this.findVNode(record.getKey());
		ArrayList<String> ar = new ArrayList<String>();
		RecordProcessor bp  = new RecordProcessor();
		ar.add("WRITERECORD");
		ar.add(this.indexName);
		String[] spo = UtilityFunctions.tupleBytesToStrings(record.getKey());
		ar.add(spo[0] + " " + spo[1] + " " + spo[2]);
		bp = (RecordProcessor)cr.SendOneVNode(new Request(vnode,ar,true),bp);
		return (bp.getData().hasNext() == false);
	}

	public List<Record> bulkAdd(List<Record> triples) {
		StringProcessor returnedData = new StringProcessor();
		HashMap<RemoteVNode, Request> reqMap = new HashMap<RemoteVNode, Request>();
		Iterator<Record> it = triples.iterator();
		while(it.hasNext()) {
			Record rec = it.next();
			String[] spo = UtilityFunctions.tupleBytesToStrings(rec.getKey());
			RemoteVNode vnode = this.findVNode(rec.getKey());
			Request r = reqMap.get(vnode); 
			
			if(r == null) {	
				r = new Request(vnode,false);
				r.addRequestLine("WRITERECORD");
				r.addRequestLine(this.indexName);
				r.addRequestLine(spo[0] + " " + spo[1] + " " + spo[2]);
				reqMap.put(vnode, r);
			}
			r.addRequestLine(spo[0] + " " + spo[1] + " " + spo[2]);
		}
		
		cr.SendMultipleRequests(reqMap.values().iterator(), returnedData);
		ArrayList<Record> duplicates = new ArrayList<Record>();
		Iterator<String> retIt = returnedData.getData();
		while(retIt.hasNext()) {
			duplicates.add(new Record(UtilityFunctions.tupleStringToByteArray(retIt.next()),null));
		}
		return duplicates;
	}
	
	public void check() {
		// TODO Auto-generated method stub

	}

	public void close() {
		NullProcessor np  = new NullProcessor();
		ArrayList<String> ar = new ArrayList<String>();
		ar.add("CLOSERINDEX");
		ar.add(this.indexName);
		cr.BroadcastAllRequest(ar, np);
	}

	public boolean contains(Record record) {
		return (this.find(record) != null);
	}

	/*
	 * Implemented as in BPlusTree
	 * @see com.hp.hpl.jena.tdb.index.Index#count()
	 */
	public long count() {
		return -1;
	}

	public boolean delete(Record record) {
		// TODO Auto-generated method stub
		return false;
	}

	/*
	 * Finds a record in the distributed index.
	 * Should send/receive key/value pairs over the network
	 * @see com.hp.hpl.jena.tdb.index.Index#find(com.hp.hpl.jena.tdb.base.record.Record)
	 */
	public Record find(Record record) {
		RemoteVNode vnode = this.findVNode(record.getKey());
		RecordProcessor rp = new RecordProcessor();
		ArrayList<String> ar = new ArrayList<String>();
		Record foundr = null;
		
		ar.add("FIND");
		ar.add(indexName);
		String str = UtilityFunctions.createSpacedTupleString(
				UtilityFunctions.tupleBytesToStrings(record.getKey()));

		ar.add(str);

		rp = (RecordProcessor)cr.SendOneVNode(new Request(vnode, ar, false), rp);
		
		Iterator<Record> tupleIterator = rp.getData();
		if(tupleIterator.hasNext()) {
			foundr = tupleIterator.next();
			//System.out.println(DataNodeServer.createSpacedTupleString(
			//		DataNodeServer.tupleBytesToStrings(foundr.getKey())));
		}
		
		return foundr;
	}

	public RecordFactory getRecordFactory() {
		return rf;
	}

	/*
	 * TODO implement as in BPlusTreeIndex
	 * @see com.hp.hpl.jena.tdb.index.Index#isEmpty()
	 */
	public boolean isEmpty() {
		// TODO Auto-generated method stub
		return false;
	}

	
	/*
	 * Currently used only for testing purposes.  Implement 
	 * at some point
	 * @see com.hp.hpl.jena.tdb.index.Index#sessionTripleCount()
	 */
	public long sessionTripleCount() {
		// TODO Auto-generated method stub
		return 0;
	}

	/*
	 * TODO send a message to all machines to sync data.
	 * @see com.hp.hpl.jena.tdb.index.Index#sync(boolean)
	 */
	public void sync(boolean force) {
		NullProcessor np  = new NullProcessor();
		ArrayList<String> ar = new ArrayList<String>();
		ar.add("SYNCRINDEX");
		ar.add(this.indexName);
		ar.add(String.valueOf(force));
		cr.BroadcastAllRequest(ar, np);
	}

	public long size() {
		// TODO Auto-generated method stub
		return 0;
	}

	

}

/*
 * (c) Copyright 2009 Hewlett-Packard Development Company, LP
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. The name of the author may not be used to endorse or promote products
 *    derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
